package com.toutiao.vssue

import scala.math.sqrt

import org.apache.log4j.Logger

import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.{HashMap => HMap}

import com.toutiao.issue.SimpleDate
import com.toutiao.issue.UpperDict

import com.toutiao.model.WAND

import com.toutiao.score.{Topic,SimScorer,SimpleFactory}
import com.toutiao.model.{Converter,TimeIndex,TimeDict,TimeDB}

import com.twitter.util.{SynchronizedLruMap => LruCache}


class VssueCollector (
        val minS: Double = 3.0,
        val mIndex:TimeIndex,
        val mTime:TimeDict,
        val mDB:TimeDB,
        val mVU: UpperDict,
        val mID: IDGetter
    ) {
  
  private
  val _log = Logger.getLogger(this.getClass.getName)
  
  def get(n:String) = {
    if (mTime.isIn(n)) {
      val t = mTime(n)
      val m = mDB(t)(n)
      m
    }
    else {
      new Topic
    }
  }
  
  private def norm(vs:Map[String,Double]) = {
    SimpleFactory.norm(vs)
  }
  
  private val MIN_SAME_TOKEN = 3
  private val MAX_SIZE_TOKEN = 35
  
  private
  def cos(m1:Map[String,Double], s1:Double, m2:Map[String,Double], s2:Double):Double = {
   SimpleFactory.cos(m1,s1,m2,s2)
  }

  private
  def merge(m1:Map[String,Double],n1:Int,m2:Map[String,Double],n2:Int) = {
    SimpleFactory.cut(SimpleFactory.merge(m1,n1,m2,n2),MAX_SIZE_TOKEN)
  }
  
  private
  def asTopic(x:Document) = {
    val e = new Topic
    e.i = x.name
    e.v = x.vs
    e.s = 1.0
    e.c = x.date
    e.u = x.date
    val ms = Array(x.name)
    e.m = Map("m"->ms)
    e
  }
  
  def getSimScore(m1:Topic,m2:Topic):Double= {
    cos(m1.v,m1.s,m2.v,m2.s)
  }
  
  def insert(name:String,i:Long,event:Topic) {
    val t = SimpleDate.format(event.c).split(" ")(0)
    mDB.insert(t, name, event)
    mIndex.insert(t, i, event.v.keySet.toArray)
    mTime.insert(name,t)
    event.v.foreach(vv => { // Update VU
      mVU.put(vv._1,vv._2.toFloat)
    })
  }
  
  def remove(t:String) = {
    mIndex.remove(t)
    mDB.remove(t)
    mTime.removeTime(t)
  }
  
  private
  def retrieve(vs:Map[String,Double],S:Double) = {
    val iis = mIndex.getIndex
    val es = iis.map(index => {
      new WAND(index,vs,S)
    })
    
    val ns = es.par.flatMap(e => e.retrieve).seq
    ns
  }
  
  private
  val CACHE = new LruCache[String,Topic](65536*16)

  def getFeature(ns:Array[String]) = {    
    val nns = ns.filter(!CACHE.contains(_))
    val xxs = ns.filter(CACHE.contains(_)).map(n => {(n,CACHE(n))}) // Hit Cache
    val xs = nns.map(n => {
      val t = mTime(n)
      (n, mDB(t).getValue(n))
    })

    val xData = xs.par.map(x => {
      val n = x._1
      val v = Converter.parseData(x._2)
      (n,v)
    }).seq.toMap

    xData.foreach( nv => { // Update Cache
    	if (nv._2.m("m").size > 1)
    	  CACHE(nv._1) = nv._2
      }
    )

    xData++xxs
  }
  
  def getData(n:String) = {
    if (CACHE.contains(n)){
      CACHE(n)
    }
    else {
      val t = mTime(n)
      val x = mDB(t).getValue(n)
      Converter.parseData(x)
    }
  }
  
  // 传出结果，Array[(String,String)] ID->IssueID
  def processOne(x:Document, S:Double) = {
    var topic = asTopic(x)
    var rs = _processOne(topic,S)
    if (rs.size == 1) {
      rs
    }
    else {
      var rrs = Set[(String,String)]()
      while (rs.size > 0) {
        //_log.info(rs.toBuffer)
        rrs ++= rs
        val rti = rs(0)._2
        topic = getData(rti)
        rs = _processOne(topic,S,rti)
      }
      rrs.toArray
    }
  }

  def _processOne(topic: Topic, S: Double, ti: String = "") = {

    // log.info("Retrieve ...")
    val ns = retrieve(
      topic.v, minS).toArray.map(_.toString)
    // log.info("Retrieve Done")

    val xData = getFeature(ns)

    val rs = ns.filter(_ != ti).par.map(n => {
      (n, getSimScore(xData(n), topic))
    }).seq

    val (z, s) = if (rs.size > 0) rs.maxBy(_._2) else ("", 0.0)

    val res =
      if (s >= S) { // 需要合并
        val t = mTime(z)
        val topicD = xData(z)
        val sizeD = topicD.m("m").size
        val vs = merge(topicD.v, sizeD, topic.v, topic.m("m").size)
        val g = topicD.s
        val ms = topicD.m("m") ++ topic.m("m") // ?

        val ct = topicD.c
        //if (topicD.c.getTime < topic.c.getTime) topicD.c
        //else topicD.c
        val ut =
          if (topicD.u.getTime < topic.u.getTime) topic.u
          else topicD.u

        val i = z.toLong // Update Index
        mIndex(t).remove(i, topicD.v.map(_._1).toArray)
        mIndex(t).insert(i, vs.map(_._1).toArray)
        
        // 删除老数据
        if (ti != "") {
          val tt = mTime(ti)
          mIndex(tt).remove(ti.toLong,topic.v.map(_._1).toArray)
          val ee = new Topic
          ee.i = topic.i
          ee.c = topic.c
          ee.u = topic.u
          ee.m = Map("m"->Array(z))
          mDB.insert(tt,ti,ee)
          _log.info("Remove "+ee.i+" "+ti)
        }
        val nn = new Topic // Update Data
        nn.i = topicD.i // 只使用被合并的
        nn.m = Map("m" -> ms.toSet.toArray)
        nn.v = vs
        nn.s = g
        nn.c = ct
        nn.u = ut
        mDB.insert(t, z, nn)
        CACHE(z) = nn // Update Cache

        topic.m("m").map(ii => { (ii, z) }).toArray :+ ((topic.i, z))
      } else {
        if (ti != "") {
          Array[(String, String)]()
        }
        else {
          val nz = mID.get
          val na = new String(nz.toString)
          insert(na, nz, topic)
          CACHE(na) = topic
          Array((topic.i, na))          
        }
      }
    //_log.info("process "+res.toBuffer)
    res
  }
  
  
}